package com.weiron.rdd.mysqlWriter;


import org.apache.spark.sql.ForeachWriter;
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema;
import java.io.Serializable;
import java.sql.*;

/**
 * @author: weiRon
 * @description mysqlWriter
 * @date: 2023/4/19 13:50
 */

public class MysqlForeachWriter extends ForeachWriter implements Serializable {

    public static Connection connection;
    public static Statement statement;


    @Override
    public boolean open(long partitionId, long version) {
        try {
            Class.forName("com.mysql.jdbc.Driver");
        } catch (ClassNotFoundException cnfEx) {

        }
        try {
            connection = DriverManager.getConnection("jdbc:mysql://127.0.0.1:3306/test", "root", "123456");
            statement = connection.createStatement();
        } catch (SQLException sqlEx) {

        }


        return true;
    }

    @Override
    public void process(Object value) {
        String qrySql = "select * from user_age_cnt where userAge='" + ((GenericRowWithSchema) value).get(0) + "'";
        // String insSql = "insert into user values ('" + ((GenericRowWithSchema) value).get(0)+"','" + ((GenericRowWithSchema) value).get(1) + "','" + ((GenericRowWithSchema) value).get(2) + "')";
        String insSql = "insert into user_age_cnt values ('" + ((GenericRowWithSchema) value).get(0) + "','" + ((GenericRowWithSchema) value).get(1) + "')";
        String updSql = "update user_age_cnt set ageCnt='" + ((GenericRowWithSchema) value).get(1) + "' where userAge='" + ((GenericRowWithSchema) value).get(0) + "'";
        try {
            PreparedStatement preStatement = connection.prepareStatement(qrySql);
            ResultSet rs = preStatement.executeQuery();
            if (rs.next()) {
                statement.execute(updSql);
                System.out.println("update SQL is: " + updSql);
            }else{
                statement.execute(insSql);
                System.out.println("update SQL is: " + insSql);
            }
        } catch (SQLException sqlEx) {

        }


    }

    @Override
    public void close(Throwable errorOrNull) {
        try {
            statement.close();
            connection.close();
        } catch (SQLException sqlEx) {

        }

    }
}